//
//  ThreadPoolCPlusPlus.cpp
//  ThreadPool
//
//  Copyright © 2022 sinosun. All rights reserved.
//

#include "ThreadPoolCPlusPlus.h"
#include "TaskQueue.hpp"
#include <string>
#include <iostream>
#include <unistd.h>
using namespace std;

const int NUMBER = 2;
ThreadPoolCPlusPlus::ThreadPoolCPlusPlus(int min, int max)
{
    // 实例化任务队列
    m_tastQ = new TaskQueue();
    do {
        // 初始化任务队列
        // 由于任务队列采用了 std::queue ，所以这里没有那些多余的队头队尾索引
        
        // 初始化线程池
        m_min = min;
        m_max = max;
        m_busyNum = 0;
        m_liveNum = min;
        m_exitNum = 0;
        
        // 根据线程池上限给工作线程ID分配内存
        m_threadIDs = new pthread_t[m_max];
        if (nullptr == m_threadIDs) {
            cout<<"malloc threadIDs[] fail..."<<endl;
            break;
        }
        m_threadIDs = {0,};
        
        // 初始化互斥锁，条件变量
        if (pthread_mutex_init(&m_lock, NULL) != 0
            || pthread_cond_init(&m_notEmpty, NULL) != 0) {
            cout<<"init mutex or cond fail..."<<endl;
            break;
        }
        
        // 创建工作者线程
        for (int i = 0; i < m_min; ++i) {
            pthread_create(&m_threadIDs[i], NULL, worker, this);
        }
        
        // 创建管理者线程
        pthread_create(&m_managerID, NULL, manager, this);
        
    } while (0);
}

ThreadPoolCPlusPlus::~ThreadPoolCPlusPlus()
{
    m_shutDown = 1;
    
    // 阻塞销毁管理者线程
    pthread_join(m_managerID, NULL);
    // 唤醒所有工作者线程
    pthread_cond_signal(&m_notEmpty);
    
    // 销毁所有锁及条件变量
    pthread_mutex_destroy(&m_lock);
    pthread_cond_destroy(&m_notEmpty);
    
    if (m_threadIDs) {
        delete[] m_threadIDs;
    }
    if (m_tastQ) {
        delete m_tastQ;
    }
}

// 添加任务
void ThreadPoolCPlusPlus::addTask(Task task)
{
    pthread_mutex_lock(&m_lock);
    m_tastQ->addTask(task);
    pthread_mutex_unlock(&m_lock);
}
// 获取活着的线程个数
int ThreadPoolCPlusPlus::getAliveNum()
{
    int aliveNum = 0;
    pthread_mutex_lock(&m_lock);
    aliveNum = m_liveNum;
    pthread_mutex_unlock(&m_lock);
    return aliveNum;
}
// 获取忙的线程个数
int ThreadPoolCPlusPlus::getBusyNum()
{
    int busyNum = 0;
    pthread_mutex_lock(&m_lock);
    busyNum = m_busyNum;
    pthread_mutex_unlock(&m_lock);
    return busyNum;
}


// 工作线程的任务函数
void* ThreadPoolCPlusPlus::worker(void* arg)
{
    ThreadPoolCPlusPlus* pool = (ThreadPoolCPlusPlus*)arg;
    while (1) {
        pthread_mutex_lock(&pool->m_lock);
        
        // 当前任务队列是否为空，为空则需要阻塞
        while (pool->m_tastQ->taskNumber() == 0 && !pool->m_shutDown )
        {
            cout << "thread" << to_string((long)pthread_self()) << "waiting..." << endl;
            pthread_cond_wait(&pool->m_notEmpty, &pool->m_lock);
            
            // 结束阻塞，判断是否需要退出线程
            if (pool->m_exitNum > 0) {
                pool->m_exitNum--;
                if (pool->getAliveNum() > pool->m_min) {
                    pool->m_liveNum--;
                    pthread_mutex_unlock(&pool->m_lock);
                    pool->threadExit();
                }
                
            }
            
            // 判断线程池是否被关闭了
            if (pool->m_shutDown) {
                pthread_mutex_unlock(&pool->m_lock);
                pool->threadExit();
            }
            
            // 从任务队列中取出一个任务
            Task task = pool->m_tastQ->takeTask();
            // 工作的线程+1
            pool->m_busyNum++;
            // 线程池解锁
            pthread_mutex_unlock(&pool->m_lock);
            // 处理任务
            
            cout << "thread " << to_string((long)pthread_self()) << " start working..." << endl;
            task.function(arg);
            delete task.arg;
            task.arg = nullptr;
            cout << "thread " << to_string((long)pthread_self()) << " end working..." << endl;
            
            // 工作完成，忙的线程数-1
            pthread_mutex_lock(&pool->m_lock);
            pool->m_busyNum--;
            pthread_mutex_unlock(&pool->m_lock);
            
            
        }
    } // end of while(1)
    return nullptr;
}
// 管理线程的任务函数
void* ThreadPoolCPlusPlus::manager(void* arg)
{
    ThreadPoolCPlusPlus* pool = static_cast<ThreadPoolCPlusPlus*>(arg);
    // 线程没有结束，就一直检测
    while (!pool->m_shutDown) {
        // 每隔5秒检查一次
        sleep(5);
        
        pthread_mutex_lock(&pool->m_lock);
        // 取出任务队列的数量
        int queueSize = pool->m_tastQ->taskNumber();
        // 取出活着的线程数
        int aliveNum = pool->getAliveNum();
        // 取出忙的线程数
        int busyNum = pool->getBusyNum();
        
        // 添加线程
        if (queueSize > aliveNum && aliveNum < pool->m_max) {
            
            pthread_mutex_lock(&pool->m_lock);
            int counter = 0;
            for (int i = 0; pool->getAliveNum() < pool->m_max
                 && counter < NUMBER
                 && i < pool->m_max; ++i) {
                if (0 == pool->m_threadIDs[i]) {
                    pthread_create(&pool->m_threadIDs[i], NULL, worker, pool);
                    counter++;
                    pool->m_liveNum++;
                }
                
            }
            pthread_mutex_unlock(&pool->m_lock);
        }
        
        // 销毁线程
        // 忙的线程的2倍 < 活着的线程数 && 活着的线程数 < max
        if (busyNum * 2 < aliveNum && aliveNum < pool->m_max) {
            pthread_mutex_lock(&pool->m_lock);
            pool->m_exitNum = NUMBER;
            pthread_mutex_unlock(&pool->m_lock);
            for (int i = 0; i < NUMBER; ++i) {
                // 唤醒阻塞的线程，让它根据 pool->m_exitNum 自杀
                pthread_cond_signal(&pool->m_notEmpty);
            }
        }
    }// end of while(1)
    return nullptr;
}
// 线程退出
void ThreadPoolCPlusPlus::threadExit()
{
    pthread_t tID = pthread_self();
    for (int i = 0; i < m_max; ++i) {
        if (m_threadIDs[i] == tID) {
            m_threadIDs[i] = 0;
            break;
        }
    }
    pthread_exit(NULL);
    cout << "thread " << to_string((long)tID) << "exit..." << endl;
}

